package com.atguigu.flink.chapter02_DataStreamAPI.sink;

import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * Created by Smexy on 2022/10/24
 *
 *     public static <T> SinkFunction<T> sink(
 *          String sql,    //SQL
 *          JdbcStatementBuilder<T> statementBuilder,  //向sql中填充占位符，填充语句
 *          JdbcExecutionOptions executionOptions,  // 执行参数
 *          JdbcConnectionOptions connectionOptions  //连接参数
 *          ) {
 */
public class Demo24_JDBCSink
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        JdbcStatementBuilder<WaterSensor> statementBuilder = new JdbcStatementBuilder<WaterSensor>()
        {
            //流中的每一个WaterSensor，执行一次
            @Override
            public void accept(PreparedStatement ps, WaterSensor waterSensor) throws SQLException {

                ps.setString(1, waterSensor.getId());
                ps.setLong(2, waterSensor.getTs());
                ps.setInt(3, waterSensor.getVc());

            }
        };

        JdbcExecutionOptions jdbcExecutionOptions = JdbcExecutionOptions.builder()
                                                         .withBatchSize(200)    //一批次最多写多少条
                                                         .withMaxRetries(3)    //失败重试的次数，最好数据库引擎支持幂等
                                                         .withBatchIntervalMs(2000)  //数据不满一批，基于时间，固定写入
                                                         .build();

        JdbcConnectionOptions jdbcConnectionOptions = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
            .withUrl("jdbc:clickhouse://hadoop102:8123/upp220608")
            .build();

        SinkFunction<WaterSensor> sink = JdbcSink.sink(
            "insert into watersensor values(?,?,?)",
            statementBuilder,
            jdbcExecutionOptions,
            jdbcConnectionOptions
        );

        env
           .socketTextStream("hadoop103", 8888)
           .map(new MapFunction<String, WaterSensor>()
           {
               @Override
               public WaterSensor map(String value) throws Exception {
                   String[] data = value.split(",");
                   return new WaterSensor(
                       data[0],
                       Long.valueOf(data[1]),
                       Integer.valueOf(data[2])
                   );
               }
           })
           .addSink(sink);

        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }
    }
}
